package cn.doitedu.producer.high.tasks;

import java.lang.reflect.Constructor;
import java.util.Iterator;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;

public class DataFetcher implements Runnable{

    Iterator<String> dataSource;
    ArrayBlockingQueue<String> queue;

    public DataFetcher(Properties properties, ArrayBlockingQueue<String> queue) throws Exception {

        // 获取数据源的配置类名
        String datasourceClassname = properties.getProperty("datasource_classname");
        // 获取数据源的配置的构造参数
        String datasourceConstructorParam = properties.getProperty("datasource_constructor_param");

        // 根据配置的数据源类名和构造参数，来反射实例对象
        Class<?> aClass = Class.forName(datasourceClassname);
        Constructor<?> constructor = aClass.getConstructor(String.class);
        dataSource = (Iterator<String>) constructor.newInstance(datasourceConstructorParam);

        this.queue = queue;
    }


    @Override
    public void run() {

        // 用数据源迭代器不断地迭代数据，放入阻塞队列
        while(dataSource.hasNext()){
            String line = dataSource.next();

            try {
                // 用put方法，（如果队列已满）有阻塞效果
                queue.put(line);

            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
}
